package com.mijie.homi.service.notify;

import java.net.InetSocketAddress;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;

import net.sf.json.JSONObject;

import org.apache.log4j.Logger;
import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.service.IoHandler;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.springframework.stereotype.Service;

import com.mijie.homi.common.constant.HomiConstants;
import com.mijie.homi.common.constant.TypeCode;
import com.mijie.homi.common.protocol.HomiProtocol;
import com.mijie.homi.common.protocol.HomiProtocolCodecFactory;
import com.mijie.util.Helper;

@Service
public class SolicitedMessageTransporter {

	private NioSocketConnector connector;
	private IoSession session = null;
	private int tryToConnectionMessageRouterCounter = -1;
	private Logger logger = Logger.getLogger(SolicitedMessageTransporter.class);
	
	private Map<String,HomiProtocol> failQueue = new HashMap<String,HomiProtocol>();
	
	private SolicitedMessageTransporter(){
		connector = new NioSocketConnector();  
		connector.getFilterChain().addLast( "codec", new ProtocolCodecFilter( new HomiProtocolCodecFactory( Charset.forName( "UTF-8" ))));
		connector.setHandler(new RouterHandler());
		connector.getSessionConfig().setKeepAlive(true);
		connectionToMessageRouter();
	}
	
	private void connectionToMessageRouter(){
		try{
			ConnectFuture future = connector.connect(new InetSocketAddress("localhost",HomiConstants.MESSAGE_ROUTER_PORT));
			future.awaitUninterruptibly();
			session = future.getSession();
		}catch(Exception e){
			logger.error(e.getMessage(), e);
			session = null;
		}
	}
	
	void send(int receiver,HomiProtocol notify){
		HomiProtocol pack = new HomiProtocol(Helper.genUUID(),TypeCode.MESSAGE_ROUTER_TYPE_SOLICITED);
		JSONObject json = new JSONObject();
		json.accumulate("receiver", receiver);
		json.accumulate("type", notify.getType());
		if(notify.getContent()!=null)
			json.accumulate("content", notify.getContent());
		pack.setContent(json.toString());
		sendMessage(pack);
	}
	
	
	private Object reConnectionLock = new Object();
	private Object flushFailQueueLock = new Object();
	private  void sendMessage(HomiProtocol pack){
		if(session==null || !session.isConnected() || session.isClosing()){
			synchronized(reConnectionLock){
				if(session==null || !session.isConnected() || session.isClosing()){
					tryToConnectionMessageRouterCounter++;
					connectionToMessageRouter();
					if(session==null && pack.getType()==TypeCode.MESSAGE_ROUTER_TYPE_SOLICITED){
						failQueue.put(pack.getId(), pack);
					}
				}
			}
		}
		if(session!=null){
			session.write(pack);
			//ignore the first time connection
			if(tryToConnectionMessageRouterCounter>0){
				tryToConnectionMessageRouterCounter = 0;
				synchronized(flushFailQueueLock){
					if(!failQueue.isEmpty()){
						Set<Map.Entry<String, HomiProtocol>> queue = failQueue.entrySet();
						Iterator<Map.Entry<String, HomiProtocol>> iter = queue.iterator();
						Map.Entry<String, HomiProtocol> entry = null;
						while(iter.hasNext()){
							entry = iter.next();
							iter.remove();
							session.write(entry.getValue());
							logger.info("##flush fail queue message->id:"+entry.getKey()+",content:"+entry.getValue().getContent());
						}
					}//end if
				}//end synchronized
			}
		}
	}
	
	class RouterHandler implements IoHandler {

		private Logger logger = Logger.getLogger(RouterHandler.class);
		
		public void exceptionCaught(IoSession session, Throwable cause)
				throws Exception {
			logger.error(cause.getMessage(), cause);
		}

		public void sessionClosed(IoSession session) throws Exception {
			logger.info("===============Disconnected with the message router===============");
		}

		public void sessionCreated(IoSession session) throws Exception {
			logger.info("===============Connected to the message router===============");
		}

		public void messageSent(IoSession session, Object message) throws Exception {
		}

		
		public void messageReceived(IoSession session, Object message)
				throws Exception {
		}

		public void sessionIdle(IoSession session, IdleStatus status)
				throws Exception {
		}

		public void sessionOpened(IoSession session) throws Exception {
			
		}
	}
}
